package spark.MLlib;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.classification.NaiveBayes;
import org.apache.spark.mllib.classification.NaiveBayesModel;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import scala.Tuple2;

/**
 * 作者: LDL
 * 功能说明:
 * 创建日期: 2015/7/1 14:19
 */
public class NaiveBayesDemo {

    public static void main(String[] args) {
        System.setProperty("hadoop.home.dir", "F:\\tools\\hadoop-common-2.2.0-bin-master");
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JavaDataTypes");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        //jsc.setLogLevel("OFF");
        JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotrans\\libsvm.txt").toJavaRDD();
        JavaRDD<LabeledPoint> test = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotest\\libsvm.txt").toJavaRDD();
        data.cache();

        /*JavaRDD<LabeledPoint> training = data.sample(false, 0.6, 10L);
        training.cache();
        JavaRDD<LabeledPoint> test = data.subtract(training);*/

        /*JavaRDD<LabeledPoint> discretizedData = test.map(
                new Function<LabeledPoint, LabeledPoint>() {
                    @Override
                    public LabeledPoint call(LabeledPoint lp) {
                        final double[] discretizedFeatures = new double[lp.features().size()];
                        for (int i = 0; i < lp.features().size(); ++i) {
                            discretizedFeatures[i] = lp.features().apply(i) / 16;
                        }
                        return new LabeledPoint(lp.label(), Vectors.dense(discretizedFeatures));
                    }
                });

        ChiSqSelector selector = new ChiSqSelector(50);
        final ChiSqSelectorModel transformer = selector.fit(discretizedData.rdd());
        JavaRDD<LabeledPoint> filteredData = discretizedData.map(
                new Function<LabeledPoint, LabeledPoint>() {
                    @Override
                    public LabeledPoint call(LabeledPoint lp) {
                        return new LabeledPoint(lp.label(), transformer.transform(lp.features()));
                    }
                }
        );*/


        final NaiveBayesModel model = NaiveBayes.train(data.rdd(), 1.0);
        //model.save(jsc.sc(),"F:/test");
        //NaiveBayesModel model = NaiveBayesModel.load(jsc.sc(), "F:/test");
        JavaPairRDD<Double, Double> predictionAndLabel =
                test.mapToPair(p -> {
                    /*System.out.println(model.predict(p.features()));
                    System.out.println(p.label());*/
                    return new Tuple2<>(model.predict(p.features()), p.label());
                });
        double accuracy = predictionAndLabel.filter(pl -> pl._1().equals(pl._2())).count() / (double) test.count();

        System.out.println(accuracy);
    }
}
